package org.apache.dolphinscheduler.service.memoryaddress;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.dolphinscheduler.common.enums.TaskType;
import org.apache.dolphinscheduler.common.utils.HadoopUtils;
import org.apache.dolphinscheduler.dao.entity.*;
import org.apache.dolphinscheduler.dao.mapper.MemoryAddressMapper;
import org.apache.dolphinscheduler.dao.mapper.ModelMapper;
import org.apache.dolphinscheduler.dao.mapper.NodeConnectionPropertyMapper;
import org.apache.dolphinscheduler.dao.mapper.NodeRelationMapper;
import org.apache.dolphinscheduler.service.exceptions.ServiceException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.Date;
import java.util.List;
import java.util.Map;

@Component
@Service
public class MemoryAddressService {

    private static final Logger logger = LoggerFactory.getLogger(MemoryAddressService.class);

    @Autowired
    MemoryAddressMapper memoryAddressMapper;

    @Autowired
    NodeConnectionPropertyMapper nodeConnectionPropertyMapper;

    @Autowired
    NodeRelationMapper nodeRelationMapper;

    @Autowired
    ModelMapper modelMapper;


    public void updateNextTag(String tag, String nextTag) {
        memoryAddressMapper.updateNextTag(tag, nextTag);
    }

    public void updateModelTag(int modelTag, int nodeid) {
        memoryAddressMapper.updateModelTag(modelTag, nodeid);
    }

    public void updateNextTagById(int id, String nextTag) {
        memoryAddressMapper.updateNextTagById(id, nextTag);
    }

    public List<MemoryAddress> getListByTag(String tag) {
        return memoryAddressMapper.getListByTag(tag);
    }

    public void insert(MemoryAddress memoryAddress) {
        memoryAddressMapper.insert(memoryAddress);
    }

    public List<MemoryAddress> getOutPathByNodeId(int nodeid) {
        return memoryAddressMapper.getOutPathByNodeId(nodeid);
    }

    public void delByID(int id) {
        memoryAddressMapper.deleteById(id);
    }

    public List<MemoryAddress> getOutPath(int nodeid) {
        return memoryAddressMapper.getOutPath(nodeid);
    }

    /**
     * 删除processInstance时 同时将memoryAddress关联数据删掉，并同时清空相关联的hdfs数据
     */
    public void delByProcessInstanceId(int processInstanceId) {

        List<MemoryAddress> memoryAddresses = memoryAddressMapper.getListByProcessId(processInstanceId);

        if (CollectionUtils.isEmpty(memoryAddresses)) {
            return;
        }

        deleteHdfs(memoryAddresses);

        memoryAddressMapper.deleteByProcessId(processInstanceId);
    }


    private void deleteHdfs(List<MemoryAddress> memoryAddresses) {
        for (int i = 0; i < memoryAddresses.size(); i++) {

            try {
                HadoopUtils.getInstance().delete(memoryAddresses.get(i).getOutPath(), true);
            } catch (IOException e) {
                logger.error(e.getMessage(), e);
                throw new ServiceException(String.format("delete resource: %s failed.", memoryAddresses.get(i).getOutPath()));
            }
        }
    }

    public void deleteByTaskInstanceId(int taskInstanceId) {
        List<MemoryAddress> memoryAddressList = memoryAddressMapper.getListByTaskInstanceId(taskInstanceId);

        deleteHdfs(memoryAddressList);

        memoryAddressMapper.deleteByNodeId(taskInstanceId);

    }


    /**
     * 将符合条件的数据插入memory address中
     * @param task task
     */
    @Transactional
    public void insertMemAddress(TaskInstance task) {
        Map<Object, String> map = task.getLastTasks();
        for (Map.Entry<Object, String> entry : map.entrySet()) {
            // 插入节点关系表，方便那个api调用
            NodeRelation nodeRelation = new NodeRelation();
            nodeRelation.setNodeId((Integer.parseInt(entry.getKey().toString())));
            nodeRelation.setNextNodeId(task.getId());
            nodeRelationMapper.insert(nodeRelation);
        }
    }

    /**
     * 将符合条件的数据插入memory address中
     */

    @Transactional
    public void insertModelMemAddress(TaskInstance task) {
        MemoryAddress memoryAddress = new MemoryAddress();
        memoryAddress.setNodeId(task.getId());
        memoryAddress.setTag(task.getSecondType());
        List<Model> list = modelMapper.getListByNames(task.getName());
        memoryAddress.setOutPath(list.get(0).getAddress());
        memoryAddress.setModelTag(2);
        memoryAddressMapper.insert(memoryAddress);
    }

    public void insertMemory(TaskInstance task, Map.Entry<Object, String> entry, MemoryAddress aList) {

        MemoryAddress memoryAddress = new MemoryAddress();
        memoryAddress.setNodeId(Integer.parseInt(entry.getKey().toString()));
        memoryAddress.setNextNodeId(task.getId());
        memoryAddress.setNextTag(task.getSecondType());
        memoryAddress.setOutPath(aList.getOutPath());
        memoryAddress.setCreateTime(new Date());
        memoryAddress.setMethodType(task.getTaskType());
        memoryAddress.setNextTagList(aList.getNextTagList());
        memoryAddress.setTag(entry.getValue());
        memoryAddress.setCreateTime(new Date());
        memoryAddress.setProcessInstanceId(task.getProcessInstanceId());
        memoryAddressMapper.deleteByNodeIdAndNextNodeId(Integer.parseInt(entry.getKey().toString()), 0);
        memoryAddressMapper.deleteByNodeIdAndNextNodeId(Integer.parseInt(entry.getKey().toString()), task.getId());
        memoryAddressMapper.insert(memoryAddress);
    }
}
